package com.example.flinkcourse.lesson3.source;

import com.example.flinkcourse.lesson3.model.Transaction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.functions.source.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Kafka 数据源工厂类
 * 用于创建 Kafka 数据源，支持 Kerberos 认证
 */
public class KafkaSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(KafkaSourceFactory.class);

    /**
     * 创建 Kafka 数据源
     * @return Kafka 数据源
     */
    public static Source<Transaction> createKafkaSource() {
        LOG.info("Creating Kafka source...");
        
        // 配置 Kafka 连接
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-group");
        
        // 配置 Kerberos 认证
        if (isKerberosEnabled()) {
            LOG.info("Kerberos authentication is enabled");
            properties.setProperty("security.protocol", "SASL_PLAINTEXT");
            properties.setProperty("sasl.mechanism", "GSSAPI");
            properties.setProperty("sasl.kerberos.service.name", "kafka");
        }
        
        // 创建 Kafka 数据源
        KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("input-topic")
            .setGroupId("flink-group")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        
        LOG.info("Kafka source created successfully");
        return source;
    }
    
    /**
     * 检查是否启用 Kerberos 认证
     * @return 是否启用 Kerberos
     */
    private static boolean isKerberosEnabled() {
        // TODO: 从配置中读取是否启用 Kerberos
        return true;
    }
    
    /**
     * 获取 Kerberos 认证参数
     * @return 认证参数
     */
    private static String getKerberosAuthParams() {
        // TODO: 从配置中读取 Kerberos 认证参数
        return "token:xxx";
    }
} 